package com.starrocks.service;

import com.starrocks.entity.SyncTableConfig;
import com.starrocks.util.IdGenerator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StreamUtils;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;

/**
 * 数据同步处理器
 * 处理INSERT、UPDATE、DELETE操作
 */
@Slf4j
@Service
public class DataSyncHandler {

    @Autowired
    @Qualifier("tidbJdbcTemplate")
    private JdbcTemplate tidbJdbcTemplate;

    @Autowired
    private DataTypeConverter dataTypeConverter;
    
    @Autowired
    private IdGenerator idGenerator;
    
    /**
     * 数据处理线程池（用于并行处理）
     */
    @Autowired
    @Qualifier("fullSyncExecutor")
    private ThreadPoolTaskExecutor dataProcessExecutor;
    
    /**
     * 并行处理配置
     */
    @Autowired
    private com.starrocks.config.ParallelProcessConfig parallelConfig;

    /**
     * 处理INSERT操作
     */
    @Transactional(rollbackFor = Exception.class)
    public void handleInsert(SyncTableConfig tableConfig, Map<String, Object> data) {
        try {
            String targetTable = tableConfig.getTargetTableName();
            
            // 🔥 1. 先过滤需要同步的字段（使用源字段名过滤）
            // 策略：如果配置了 syncFields，则只保留这些字段；如果没配置，保留所有字段
            Map<String, Object> filteredData = data;
            if (tableConfig.getSyncFields() != null && !tableConfig.getSyncFields().isEmpty()) {
                filteredData = new HashMap<>();
                for (Map.Entry<String, Object> entry : data.entrySet()) {
                    String sourceField = entry.getKey();
                    // 只保留在同步字段列表中的字段（主键必须保留）
                    if (tableConfig.getSyncFields().contains(sourceField) 
                        || sourceField.equals(tableConfig.getPrimaryKey())) {
                        filteredData.put(sourceField, entry.getValue());
                    }
                }
                log.debug("INSERT - 过滤后字段数: {} (原始: {}), syncFields: {}", 
                    filteredData.size(), data.size(), tableConfig.getSyncFields().size());
            }
            
            // 2. 数据类型转换
            Map<String, Object> convertedData = dataTypeConverter.convertRowData(filteredData, targetTable);
            
            // 3. 应用字段映射（StarRocks字段名 → TiDB字段名）
            Map<String, Object> mappedData = tableConfig.applyFieldMapping(convertedData);
            
            // 4. 处理ID生成策略
            if (tableConfig.getUseTargetId() != null && tableConfig.getUseTargetId()) {
                String targetPrimaryPrimaryKey = tableConfig.getTargetPrimaryPrimaryKey();
                Object generatedId = idGenerator.generateId(tableConfig.getIdGenerateStrategy());
                if (generatedId != null) {
                    // 添加目标表自用id
                    mappedData.put(targetPrimaryPrimaryKey, generatedId);
                    log.debug("使用生成的ID: {}, 策略: {}", generatedId, tableConfig.getIdGenerateStrategy());
                } else {
                    // 如果是AUTO_INCREMENT，移除主键字段，让数据库自动生成
                    if ("AUTO_INCREMENT".equalsIgnoreCase(tableConfig.getIdGenerateStrategy())) {
                        mappedData.remove(targetPrimaryPrimaryKey);
                        log.debug("使用数据库AUTO_INCREMENT生成ID");
                    }
                }
            }
            
            // 5. 获取所有要插入的字段（已经是目标字段名，已经过滤过）
            List<String> columns = new ArrayList<>(mappedData.keySet());
            
            if (columns.isEmpty()) {
                log.warn("表 {} 没有可同步的字段", targetTable);
                return;
            }
            
            // 构建INSERT语句
            String columnStr = String.join(", ", columns);
            String placeholders = columns.stream().map(c -> "?").collect(Collectors.joining(", "));
            String insertSql = String.format("INSERT INTO %s (%s) VALUES (%s)", 
                targetTable, columnStr, placeholders);
            
            // 提取值（已转换和映射）
            Object[] values = columns.stream().map(mappedData::get).toArray();
            
            // 执行插入
            tidbJdbcTemplate.update(insertSql, values);
            log.info("INSERT成功 - 表: {}, 主键: {}, 插入字段数: {}", targetTable, 
                mappedData.get(tableConfig.getTargetPrimaryKey()), columns.size());
            
        } catch (Exception e) {
            log.error("INSERT失败 - 表: {}, 数据: {}", tableConfig.getTargetTableName(), data, e);
            throw new RuntimeException("INSERT操作失败", e);
        }
    }

    /**
     * 处理UPDATE操作
     */
    @Transactional(rollbackFor = Exception.class)
    public void handleUpdate(SyncTableConfig tableConfig, Map<String, Object> data) {
        try {
            String targetTable = tableConfig.getTargetTableName();
            String targetPrimaryKey = tableConfig.getTargetPrimaryKey();
            
            // 🔥 1. 先过滤需要同步的字段（使用源字段名过滤）
            // 策略：如果配置了 syncFields，则只保留这些字段；如果没配置，保留所有字段
            Map<String, Object> filteredData = data;
            if (tableConfig.getSyncFields() != null && !tableConfig.getSyncFields().isEmpty()) {
                filteredData = new HashMap<>();
                for (Map.Entry<String, Object> entry : data.entrySet()) {
                    String sourceField = entry.getKey();
                    // 只保留在同步字段列表中的字段（主键必须保留）
                    if (tableConfig.getSyncFields().contains(sourceField) 
                        || sourceField.equals(tableConfig.getPrimaryKey())) {
                        filteredData.put(sourceField, entry.getValue());
                    }
                }
                log.debug("UPDATE - 过滤后字段数: {} (原始: {}), syncFields: {}", 
                    filteredData.size(), data.size(), tableConfig.getSyncFields().size());
            }
            
            // 2. 检查源数据是否被软删除（在转换前检查，使用原始数据）
            boolean isSourceDeleted = isSourceDataDeleted(data, tableConfig);
            
            // 3. 如果目标表不使用软删除，且源数据被软删除，则物理删除目标表数据
            if (!tableConfig.getUseSoftDelete() && isSourceDeleted) {
                // 先获取主键值用于删除
                Map<String, Object> convertedData = dataTypeConverter.convertRowData(filteredData, targetTable);
                Map<String, Object> mappedData = tableConfig.applyFieldMapping(convertedData);
                Object primaryKeyValue = mappedData.get(targetPrimaryKey);
                
                if (primaryKeyValue != null) {
                    log.info("源数据已被软删除，且目标表不使用软删除，执行物理删除 - 表: {}, 主键: {}", targetTable, primaryKeyValue);
                    handlePhysicalDelete(tableConfig, primaryKeyValue);
                } else {
                    log.warn("源数据已被软删除，但无法获取主键值，跳过删除 - 表: {}", targetTable);
                }
                return;
            }
            
            // 4. 数据类型转换
            Map<String, Object> convertedData = dataTypeConverter.convertRowData(filteredData, targetTable);
            
            // 5. 应用字段映射（StarRocks字段名 → TiDB字段名）
            Map<String, Object> mappedData = tableConfig.applyFieldMapping(convertedData);
            
            Object primaryKeyValue = mappedData.get(targetPrimaryKey);
            
            if (primaryKeyValue == null) {
                log.error("UPDATE失败 - 主键值为空，表: {}", targetTable);
                return;
            }
            
            // 6. 正常更新流程
            // 获取要更新的字段（已经是目标字段名，已经过滤过）
            List<String> columns = new ArrayList<>(mappedData.keySet());
//            columns.remove(targetPrimaryKey); // 移除主键

            if (columns.isEmpty()) {
                log.warn("表 {} 没有可更新的字段", targetTable);
                return;
            }
            
            // 构建UPDATE语句
            String setClause = columns.stream()
                .map(col -> col + " = ?")
                .collect(Collectors.joining(", "));

            String updateSql;
            if (tableConfig.getUseTargetId()){
                String primaryKey = tableConfig.getPrimaryKey();
                if (!tableConfig.getSyncFields().isEmpty()){
                    if (tableConfig.getSyncFields().contains(primaryKey)) {
                        primaryKey = tableConfig.getFieldMapping().get(primaryKey);
                    }
                }
                if (!StringUtils.hasText(primaryKey)){
                    throw new RuntimeException("使用了目标表自用主键但是找不到映射关系！");
                }
                updateSql = String.format("UPDATE %s SET %s WHERE %s = ?",
                        targetTable, setClause, primaryKey);
            }else {
                updateSql = String.format("UPDATE %s SET %s WHERE %s = ?",
                        targetTable, setClause, tableConfig.getPrimaryKey());
            }

            
            // 提取值（已转换和映射）
            List<Object> values = new ArrayList<>();
            columns.forEach(col -> values.add(mappedData.get(col)));
            values.add(primaryKeyValue);
            
            // 执行更新
            int affected = tidbJdbcTemplate.update(updateSql, values.toArray());
            
            if (affected > 0) {
                log.info("UPDATE成功 - 表: {}, 主键: {}, 影响行数: {}, 更新字段数: {}", 
                    targetTable, primaryKeyValue, affected, columns.size());
            } else {
                log.warn("UPDATE未影响任何行 - 表: {}, 主键: {}，尝试INSERT", targetTable, primaryKeyValue);
                // 如果没有更新到数据，可能是目标表中不存在该记录，尝试插入
                handleInsert(tableConfig, data);
            }
            
        } catch (Exception e) {
            log.error("UPDATE失败 - 表: {}, 数据: {}", tableConfig.getTargetTableName(), data, e);
            throw new RuntimeException("UPDATE操作失败", e);
        }
    }
    
    /**
     * 检查源数据是否被软删除
     * 使用配置的软删除字段名进行检查
     */
    private boolean isSourceDataDeleted(Map<String, Object> data, SyncTableConfig tableConfig) {
        // 获取配置的软删除字段名，默认为 "is_delete"
        String deleteField = tableConfig.getDeleteField() != null ? tableConfig.getDeleteField() : "is_delete";
        
        // 检查软删除字段（1表示删除，0或NULL表示正常）
        Object deleteValue = data.get(deleteField);
        if (deleteValue != null) {
            if (deleteValue instanceof Number) {
                return ((Number) deleteValue).intValue() == 1;
            } else if (deleteValue instanceof Boolean) {
                return (Boolean) deleteValue;
            } else if (deleteValue instanceof String) {
                return "1".equals(deleteValue) || "true".equalsIgnoreCase((String) deleteValue);
            }
        }
        
        return false; // 默认为未删除
    }
    
    /**
     * 物理删除目标表数据
     */
    @Transactional(rollbackFor = Exception.class)
    public void handlePhysicalDelete(SyncTableConfig tableConfig, Object primaryKeyValue) {
        try {
            String targetTable = tableConfig.getTargetTableName();
            String targetPrimaryKey = tableConfig.getTargetPrimaryKey();
            
            // 构建DELETE语句
            String deleteSql = String.format("DELETE FROM %s WHERE %s = ?", targetTable, targetPrimaryKey);
            
            // 执行删除
            int affected = tidbJdbcTemplate.update(deleteSql, primaryKeyValue);
            log.info("物理删除成功 - 表: {}, 主键: {}, 影响行数: {}", targetTable, primaryKeyValue, affected);
            
        } catch (Exception e) {
            log.error("物理删除失败 - 表: {}, 主键: {}", tableConfig.getTargetTableName(), primaryKeyValue, e);
            throw new RuntimeException("物理删除操作失败", e);
        }
    }

    /**
     * 处理DELETE操作
     */
    @Transactional(rollbackFor = Exception.class)
    public void handleDelete(SyncTableConfig tableConfig, Map<String, Object> data) {
        try {
            String targetTable = tableConfig.getTargetTableName();
            String targetPrimaryKey = tableConfig.getTargetPrimaryKey();
            
            // 应用字段映射（虽然DELETE通常只需要主键）
            Map<String, Object> mappedData = tableConfig.applyFieldMapping(data);
            Object primaryKeyValue = mappedData.get(targetPrimaryKey);
            
            if (primaryKeyValue == null) {
                log.error("DELETE失败 - 主键值为空，表: {}", targetTable);
                return;
            }
            
            // 构建DELETE语句
            String deleteSql = String.format("DELETE FROM %s WHERE %s = ?", targetTable, targetPrimaryKey);
            
            // 执行删除
            int affected = tidbJdbcTemplate.update(deleteSql, primaryKeyValue);
            log.info("DELETE成功 - 表: {}, 主键: {}, 影响行数: {}", targetTable, primaryKeyValue, affected);
            
        } catch (Exception e) {
            log.error("DELETE失败 - 表: {}, 数据: {}", tableConfig.getTargetTableName(), data, e);
            throw new RuntimeException("DELETE操作失败", e);
        }
    }

    /**
     * 清空表（用于全量同步）
     */
    @Transactional(rollbackFor = Exception.class)
    public void truncateTable(SyncTableConfig tableConfig) {
        try {
            String targetTable = tableConfig.getTargetTableName();
            String truncateSql = String.format("TRUNCATE TABLE %s", targetTable);
            
            tidbJdbcTemplate.execute(truncateSql);
            log.info("成功清空表: {}", targetTable);
            
        } catch (Exception e) {
            log.error("清空表 {} 失败", tableConfig.getTargetTableName(), e);
            throw new RuntimeException("清空表操作失败", e);
        }
    }

    /**
     * 批量插入（优化版：大数据量时使用并行处理）
     */
    @Transactional(rollbackFor = Exception.class)
    public void batchInsert(SyncTableConfig tableConfig, List<Map<String, Object>> dataList) {
        if (dataList == null || dataList.isEmpty()) {
            return;
        }
        
        try {
            String targetTable = tableConfig.getTargetTableName();
            // 根据配置和数据量决定是否使用并行处理
            boolean useParallel = parallelConfig.isEnabled() && dataList.size() >= parallelConfig.getThreshold();
            long startTime = System.currentTimeMillis();
            
            // 🔥 1. 先过滤需要同步的字段（使用源字段名过滤）
            final List<Map<String, Object>> filteredDataList = performFiltering(
                dataList, tableConfig, useParallel);
            
            log.debug("批量INSERT - 过滤完成, 数据量: {} -> {}, 处理方式: {}", 
                dataList.size(), filteredDataList.size(), useParallel ? "并行" : "串行");
            
            long filterTime = System.currentTimeMillis() - startTime;
            
            // 2. 批量转换数据类型和应用字段映射（并行处理）
            final List<Map<String, Object>> processedDataList = performConversion(
                filteredDataList, tableConfig, targetTable, useParallel);
            
            long convertTime = System.currentTimeMillis() - startTime - filterTime;
            
            // 3. 获取所有要插入的字段（已经是目标字段名，已经过滤过）
            Map<String, Object> firstRow = processedDataList.get(0);
            List<String> columns = new ArrayList<>(firstRow.keySet());
            
            if (columns.isEmpty()) {
                log.warn("表 {} 没有可同步的字段", targetTable);
                return;
            }
            
            // 构建INSERT语句
            String columnStr = String.join(", ", columns);
            String placeholders = columns.stream().map(c -> "?").collect(Collectors.joining(", "));
            String insertSql = String.format("INSERT INTO %s (%s) VALUES (%s)", 
                targetTable, columnStr, placeholders);
            
            // 批量执行（已转换和映射）- 数据库操作保持串行
            List<Object[]> batchArgs = processedDataList.stream()
                .map(data -> columns.stream().map(data::get).toArray())
                .collect(Collectors.toList());
            
            long prepareTime = System.currentTimeMillis() - startTime - filterTime - convertTime;
            long insertStartTime = System.currentTimeMillis();
            
            tidbJdbcTemplate.batchUpdate(insertSql, batchArgs);
            
            long insertTime = System.currentTimeMillis() - insertStartTime;
            long totalTime = System.currentTimeMillis() - startTime;
            
            log.info("批量INSERT成功 - 表: {}, 记录数: {}, 插入字段数: {}, 处理方式: {}, " +
                    "耗时分布[过滤:{}ms, 转换:{}ms, 准备:{}ms, 插入:{}ms, 总计:{}ms]", 
                targetTable, dataList.size(), columns.size(), useParallel ? "并行" : "串行",
                filterTime, convertTime, prepareTime, insertTime, totalTime);
            
        } catch (Exception e) {
            log.error("批量INSERT失败 - 表: {}", tableConfig.getTargetTableName(), e);
            throw new RuntimeException("批量INSERT操作失败", e);
        }
    }
    
    /**
     * 过滤字段（提取为独立方法，便于并行调用）
     * 策略：如果配置了 syncFields，只保留这些字段；如果没配置，保留所有字段
     */
    private Map<String, Object> filterFields(Map<String, Object> data, SyncTableConfig tableConfig) {
        Map<String, Object> filteredData = new HashMap<>();
        for (Map.Entry<String, Object> entry : data.entrySet()) {
            String sourceField = entry.getKey();
            // 只保留在同步字段列表中的字段（主键必须保留）
            if (tableConfig.getSyncFields().contains(sourceField) 
                || sourceField.equals(tableConfig.getPrimaryKey())) {
                filteredData.put(sourceField, entry.getValue());
            }
        }
        return filteredData;
    }
    
    /**
     * 执行字段过滤（支持并行/串行）
     */
    private List<Map<String, Object>> performFiltering(
            List<Map<String, Object>> dataList, 
            SyncTableConfig tableConfig, 
            boolean useParallel) {
        
        // 如果没有配置同步字段，直接返回原始数据
        if (tableConfig.getSyncFields() == null || tableConfig.getSyncFields().isEmpty()) {
            return dataList;
        }
        
        // 根据是否并行处理选择不同的处理方式
        if (useParallel) {
            // 使用自定义线程池的并行流
            ForkJoinPool customPool = createCustomForkJoinPool();
            try {
                return customPool.submit(() ->
                    dataList.parallelStream()
                        .map(data -> filterFields(data, tableConfig))
                        .collect(Collectors.toList())
                ).get();
            } catch (Exception e) {
                log.warn("并行过滤失败，回退到串行处理", e);
                return dataList.stream()
                    .map(data -> filterFields(data, tableConfig))
                    .collect(Collectors.toList());
            } finally {
                customPool.shutdown();
            }
        } else {
            // 小数据量，串行处理
            return dataList.stream()
                .map(data -> filterFields(data, tableConfig))
                .collect(Collectors.toList());
        }
    }
    
    /**
     * 执行数据类型转换和字段映射（支持并行/串行）
     */
    private List<Map<String, Object>> performConversion(
            List<Map<String, Object>> dataList,
            SyncTableConfig tableConfig,
            String targetTable,
            boolean useParallel) {
        
        if (useParallel) {
            // 使用自定义线程池的并行流
            ForkJoinPool customPool = createCustomForkJoinPool();
            try {
                return customPool.submit(() ->
                    dataList.parallelStream()
                        .map(data -> dataTypeConverter.convertRowData(data, targetTable))
                        .map(data -> tableConfig.applyFieldMapping(data))
                        .collect(Collectors.toList())
                ).get();
            } catch (Exception e) {
                log.warn("并行转换失败，回退到串行处理", e);
                return dataList.stream()
                    .map(data -> dataTypeConverter.convertRowData(data, targetTable))
                    .map(data -> tableConfig.applyFieldMapping(data))
                    .collect(Collectors.toList());
            } finally {
                customPool.shutdown();
            }
        } else {
            // 小数据量，串行处理
            return dataList.stream()
                .map(data -> dataTypeConverter.convertRowData(data, targetTable))
                .map(data -> tableConfig.applyFieldMapping(data))
                .collect(Collectors.toList());
        }
    }
    
    /**
     * 创建自定义的 ForkJoinPool
     * 使用线程池的最大线程数作为并行度
     */
    private ForkJoinPool createCustomForkJoinPool() {
        int parallelism = dataProcessExecutor.getMaxPoolSize();
        return new ForkJoinPool(parallelism);
    }
}

